package cn.edu.flink.scala.tutorial.sql

import cn.edu.flink.scala.tutorial.source.ClickSource
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.bridge.scala._

object GroupTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val clickStream = env.addSource(new ClickSource)

    val tEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    tEnv.createTemporaryView("click_table", clickStream)

    val maryClickTable = tEnv.sqlQuery(
      s"""
         |select user, url
         |from click_table
         |where user = 'Mary'
         |""".stripMargin)

    val maryClickCountTable = tEnv.sqlQuery(
      s"""
         |select user, count(url) click_count
         |from ${maryClickTable}
         |group by user
         |""".stripMargin)

    // 聚合结果需要转换成修改日志流
//    tEnv.toChangelogStream(maryClickTable).print("mary click")
//    tEnv.toChangelogStream(maryClickCountTable).print("mary click count")

    maryClickCountTable.execute().print()

    env.execute("GroupTest")
  }
}
